package com.exercise.rpc.client;/**
 * Created by kete on 2018-10-26.
 */

import com.exercise.rpc.common.model.RpcRequest;
import com.exercise.rpc.common.model.RpcResponse;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

/**
 * 作者：kete
 * 创建时间：2018-10-26 15:19
 * 功能描述：
 * 版本：
 */
public class ClientHandler extends SimpleChannelInboundHandler<RpcResponse> {
    private static final Logger logger = LoggerFactory.getLogger(ClientHandler.class);

    private volatile Channel channel;

    private SocketAddress remotePeer;

    private ConcurrentHashMap<String, RpcFuture> pendingRPC = new ConcurrentHashMap<>();

    public SocketAddress getRemotePeer() {
        return remotePeer;
    }


    public Channel getChannel() {
        return channel;
    }


    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel = ctx.channel();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.remotePeer = this.channel.remoteAddress();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) throws Exception {
        String requestId = msg.getRequestId();
        logger.debug("RpcResponse:"+requestId);
        RpcFuture future = pendingRPC.get(requestId);
        if(future!=null){
            pendingRPC.remove(requestId);
            future.done(msg);
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("api caught exception", cause);
        ctx.close();
    }

    public void close(){
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }

    public RpcFuture sendMsg(RpcRequest request){
        final CountDownLatch countDownLatch = new CountDownLatch(1);

        RpcFuture future = new RpcFuture(request);

        pendingRPC.put(request.getRequestId(),future);

        channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                countDownLatch.countDown();
            }
        });

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return future;

    }
}
